package com.sgx.nba

import org.apache.spark.sql.{SparkSession, functions => F}

object NBABSAnalyticsApp {
  def main(args: Array[String]): Unit = {
    // 初始化 SparkSession
    val spark = SparkSession.builder()
      .appName("球队战绩分析")
      .master("local[*]") // 使用 Local 模式
      .config("spark.sql.warehouse.dir", "spark-warehouse")
      .getOrCreate()

    // 加载 CSV 数据
    val filePath = "data/zhanjipaihang.csv"
    val df = spark.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv(filePath)

    // 数据探索
    df.printSchema()
    df.show(5)

    // MySQL 配置
    val jdbcUrl = "jdbc:mysql://localhost:3306/springbootf26x9x94?useSSL=false&serverTimezone=UTC"
    val jdbcUser = "root"
    val jdbcPassword = "123456"

    // 1. 按赛区分组，计算各赛区的平均得分，并写入 MySQL
    val avgScoreBySaiqu = df.groupBy("saiqu")
      .agg(F.avg("defen").alias("平均得分"))
      .orderBy(F.desc("平均得分"))
    avgScoreBySaiqu.write
      .format("jdbc")
      .option("url", jdbcUrl)
      .option("dbtable", "avg_score_by_saiqu")
      .option("user", jdbcUser)
      .option("password", jdbcPassword)
      .mode("overwrite")
      .save()

    // 2. 胜率最高的队伍，写入 MySQL
    val topWinRateTeams = df.orderBy(F.desc("shenglv"))
      .select("duiming", "shenglv")
      .limit(10)
    topWinRateTeams.write
      .format("jdbc")
      .option("url", jdbcUrl)
      .option("dbtable", "top_win_rate_teams")
      .option("user", jdbcUser)
      .option("password", jdbcPassword)
      .mode("overwrite")
      .save()

    // 3. 净胜分最高的队伍，写入 MySQL
    val topNetWinTeams = df.orderBy(F.desc("jingsheng"))
      .select("duiming", "jingsheng")
      .limit(10)
    topNetWinTeams.write
      .format("jdbc")
      .option("url", jdbcUrl)
      .option("dbtable", "top_net_win_teams")
      .option("user", jdbcUser)
      .option("password", jdbcPassword)
      .mode("overwrite")
      .save()

    // 4. 胜场差排名，写入 MySQL
    val winDifferenceRanking = df.orderBy(F.desc("scc"))
      .select("duiming", "scc")
      .limit(10)
    winDifferenceRanking.write
      .format("jdbc")
      .option("url", jdbcUrl)
      .option("dbtable", "win_difference_ranking")
      .option("user", jdbcUser)
      .option("password", jdbcPassword)
      .mode("overwrite")
      .save()

    // 5. 得失分差计算结果，写入 MySQL
    val scoreDifference = df.withColumn("得失分差", F.col("defen") - F.col("shifen"))
      .select("duiming", "得失分差")
      .orderBy(F.desc("得失分差"))
    scoreDifference.write
      .format("jdbc")
      .option("url", jdbcUrl)
      .option("dbtable", "score_difference")
      .option("user", jdbcUser)
      .option("password", jdbcPassword)
      .mode("overwrite")
      .save()

    println("数据成功写入 MySQL！")
  }
}
